package com.chenjj.bigdata.zk.leaderelection;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.util.Enumeration;

public class Node {
    private Status status;
    private String nodeName;
    private String path;
    private ZkClient zkClient;
    private Business business;
    private static Logger logger = LoggerFactory.getLogger(Node.class);
    /**
     * @param path zk的node path,如/test
     * @param zkAddress zookeeper的地址,如127.0.0.1:2181
     * @param sesstionOut zkClient的session超时时间
     * @throws IOException
     */
    public Node(String path,String zkAddress,int sesstionOut ) throws IOException {
        this.path = path;
        this.zkClient = new ZkClient(new ZkConnection(zkAddress),sesstionOut);
        this.nodeName = getIpAddress() + ":" + Thread.currentThread().getName();
        //添加监听
        zkClient.subscribeDataChanges(this.path, new IZkDataListener() {
            public void handleDataChange(String dataPath, Object data) throws Exception {
                logger.debug("data of path {} hasChanged, new data:{}",dataPath,data);
            }

            public void handleDataDeleted(String dataPath) throws Exception {
                logger.info("data of path {} has been deleted, ready to lookingForLeader",dataPath);
                lookingForLeader();
            }
        });
        lookingForLeader();
    }
    public void lookingForLeader() {
        String nodeInfo = this.nodeName;
        status = Status.LOOKING;
        try {
            //创建临时节点
            zkClient.createEphemeral(path,nodeInfo);
            // 如果上一步没有抛异常，说明自己已经是leader了
            status = Status.LEADER;
            logger.info("{} is leader",nodeInfo);
        } catch (ZkNodeExistsException e) {
            // 节点已经存在，说明leader已经被别人注册成功了，自己是follower
            status = Status.FOLLOWER;
            logger.info("{} is follower",nodeInfo);
        }
    }
    public void shutdown() {
        if (zkClient != null) {
            zkClient.close();
        }
    }

    public String getNodeName(){ return nodeName; }

    public Status getStatus() {
        return status;
    }
    // 当前节点的状态，节点的状态必须在这三个中的一个
    public enum Status {
        LOOKING, // 选举中
        LEADER, // 选举完毕，当前节点为leader
        FOLLOWER; // 选举完毕，当前节点为follower
    }

    /**
     * 获取本机ip地址
     * @return
     */
    public static String getIpAddress() {
        try {
            Enumeration<NetworkInterface> allNetInterfaces = NetworkInterface.getNetworkInterfaces();
            InetAddress ip = null;
            while (allNetInterfaces.hasMoreElements()) {
                NetworkInterface netInterface = (NetworkInterface) allNetInterfaces.nextElement();
                if (netInterface.isLoopback() || netInterface.isVirtual() || !netInterface.isUp()) {
                    continue;
                } else {
                    Enumeration<InetAddress> addresses = netInterface.getInetAddresses();
                    while (addresses.hasMoreElements()) {
                        ip = addresses.nextElement();
                        if (ip != null && ip instanceof Inet4Address) {
                            return ip.getHostAddress();
                        }
                    }
                }
            }
        } catch (Exception e) {
            System.err.println("IP地址获取失败" + e.toString());
        }
        return "not found";
    }
}
